package com.producer.service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

@Component
@Slf4j
public class KafkaSender {
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Async
    public void send() {
        System.out.println("异步发送消息");
        Map<String, String> messageMap = new HashMap<>();
        messageMap.put("log", "日志：执行完成");
        ObjectMapper objectMapper = new ObjectMapper();
        String data;
        try {
            data = objectMapper.writeValueAsString(messageMap);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
        String key = String.valueOf(UUID.randomUUID());
        //kakfa的推送消息方法有多种，可以采取带有任务key的，也可以采取不带有的（不带时默认为null）
        this.send("testTopic", key, data);
    }

    public void send(String topic, String key, String data) {
        //发送消息
        CompletableFuture<SendResult<String, Object>> completable = kafkaTemplate.send(topic, key, data);
        completable.whenCompleteAsync((result, ex) -> {
            if (null == ex) {
                log.info(topic + "生产者发送消息成功：" + result.toString());
            } else {
                log.info(topic + "生产者发送消息失败：" + ex.getMessage());
            }
        });
    }
}